#include "pthread.h"
#include "stdio.h"
#include "shm.h"
#include "common.h"
#include "msg_handler.h"
//#include <sys/time.h>
#include <unistd.h>
#include <string.h>


MsgQueue queue;                  // 消息队列实例
MQTTClient pub_client = NULL;    // MQTT发布客户端
MQTTClient sub_client = NULL;    // MQTT订阅客户端
void* shm_addr =NULL;    // 共享内存地址
int g_quit = 0;                  // 程序退出标志

// 线程1：读共享内存
void* read_shm_thread(void* arg) {
    devdata data;
    while (!g_quit) {
        if (shm_read(shm_addr, &data) != 0) {
            printf("[read_shm_thread] 共享内存读取失败,1秒后重试\n");
            sleep(1);
            continue;
        }

    char payload[256];
    snprintf(payload, sizeof(payload), 
         "dev_id=%d,tem=%.2f,hum=%.2f",
         data.dev_id, data.tem, data.hum);

// 发布字符串消息
if (mqtt_publish(pub_client, payload) != 0) {
    printf("MQTT发布失败\n");
} else {
    printf("MQTT发布成功:%s\n", payload);
}
    sleep(1);
  }

    printf("[read_shm_thread] 退出\n");
    return NULL;
}

// 线程2：MQTT发布线程
void* mqtt_publish_thread(void* arg) {
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    int rc;

    if ((rc = MQTTClient_create(&pub_client, ADDRESS, PUB_CLIENT,
        MQTTCLIENT_PERSISTENCE_NONE, NULL)) != MQTTCLIENT_SUCCESS) {
        printf("[mqtt_publish_thread] 创建客户端失败，错误码: %d\n", rc);
        g_quit = 1;
        return NULL;
    }

    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;
    if ((rc = MQTTClient_connect(pub_client, &conn_opts)) != MQTTCLIENT_SUCCESS) {
        printf("[mqtt_publish_thread] 连接失败，错误码: %d\n", rc);
        MQTTClient_destroy(&pub_client);
        g_quit = 1;
        return NULL;
    }
    printf("[mqtt_publish_thread] 发布客户端连接成功（主题：%s)\n", PUB_TOPIC);

    while (!g_quit) sleep(1);

    MQTTClient_disconnect(pub_client, TIMEOUT);
    MQTTClient_destroy(&pub_client);
    printf("[mqtt_publish_thread] 退出\n");
    return NULL;
}

// MQTT消息到达回调（供订阅线程使用）
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message) {
    if (g_quit) return 1;

    char* msg_json = (char*)malloc(message->payloadlen + 1);
    memcpy(msg_json, message->payload, message->payloadlen);
    msg_json[message->payloadlen] = '\0';

    msg_queue_send(&queue, msg_json);
    printf("[msgarrvd] QT指令存入队列:%s(主题：%s)\n", msg_json, topicName);

    free(msg_json);
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

// 线程3：MQTT订阅线程
void* mqtt_subscribe_thread(void* arg) {
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    int rc;

    if ((rc = MQTTClient_create(&sub_client, ADDRESS, SUB_CLIENT,
        MQTTCLIENT_PERSISTENCE_NONE, NULL)) != MQTTCLIENT_SUCCESS) {
        printf("[mqtt_subscribe_thread] 创建客户端失败，错误码: %d\n", rc);
        g_quit = 1;
        return NULL;
    }

    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;
    MQTTClient_setCallbacks(sub_client, NULL, connlost, msgarrvd, NULL);

    if ((rc = MQTTClient_connect(sub_client, &conn_opts)) != MQTTCLIENT_SUCCESS) {
        printf("[mqtt_subscribe_thread] 连接失败，错误码: %d\n", rc);
        MQTTClient_destroy(&sub_client);
        g_quit = 1;
        return NULL;
    }
    if ((rc = MQTTClient_subscribe(sub_client, SUB_TOPIC, QOS)) != MQTTCLIENT_SUCCESS) {
        printf("[mqtt_subscribe_thread] 订阅失败，错误码: %d\n", rc);
        MQTTClient_disconnect(sub_client, TIMEOUT);
        MQTTClient_destroy(&sub_client);
        g_quit = 1;
        return NULL;
    }
    printf("[mqtt_subscribe_thread] 订阅客户端连接成功（主题：%s)\n", SUB_TOPIC);

    while (!g_quit) sleep(1);

    MQTTClient_unsubscribe(sub_client, SUB_TOPIC);
    MQTTClient_disconnect(sub_client, TIMEOUT);
    MQTTClient_destroy(&sub_client);
    printf("[mqtt_subscribe_thread] 退出\n");
    return NULL;
}

// 线程4：消息队列管理线程
void* queue_thread(void* arg) {
    msg_queue_init(&queue);
    printf("[queue_thread] 消息队列初始化完成(等待QT指令)\n");

    while (!g_quit) {
        char* msg = msg_queue_read(&queue);  // 阻塞读取消息
        if (msg) {
              free(msg);  // 释放消息内存
        } 
        if (g_quit) {
            break;
        }
    }

    msg_queue_destory(&queue);
    printf("[queue_thread] 消息队列销毁，退出\n");
    return NULL;
}




// 统一创建所有线程
int create_device_threads(pthread_t* t1, pthread_t* t2, pthread_t* t3, pthread_t* t4) {
    if (pthread_create(t1, NULL, read_shm_thread, NULL) != 0) {
        printf("[create_all_threads] 创建read_shm_thread失败\n");
        return -1;
    }
    if (pthread_create(t2, NULL, mqtt_publish_thread, NULL) != 0) {
        printf("[create_all_threads] 创建mqtt_publish_thread失败\n");
        return -1;
    }
    if (pthread_create(t3, NULL, mqtt_subscribe_thread, NULL) != 0) {
        printf("[create_all_threads] 创建mqtt_subscribe_thread失败\n");
        return -1;
    }
    if (pthread_create(t4, NULL, queue_thread, NULL) != 0) {
        printf("[create_all_threads] 创建queue_thread失败\n");
        return -1;
    }
    return 0;
}




